package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo18Action {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    val stuRDD: RDD[String] = sc.textFile("scala/data/stu/students.txt")

    /**
     * 常用的行为算子：foreach、reduce、take、collect、collectAsMap、reduceByKeyLocally、top、lookup
     */

    // 打印出id和name
    stuRDD.map(line => {
      val splits: Array[String] = line.split(",")
      s"${splits(0)},${splits(1)}"
    }).foreach(println)

    stuRDD.foreach(line => {
      val splits: Array[String] = line.split(",")
      println(s"${splits(0)},${splits(1)}")
    })

    // reduce：全局的聚合
    val sumAge: Int = stuRDD.map(line => line.split(",")(2).toInt).reduce(_ + _)
    println(sumAge)

    // count：统计RDD中的数据的条数
    val cnt: Long = stuRDD.count()
    println(s"平均年龄：${sumAge * 1.0 / cnt}")

    // take：可以取出RDD的N条数据，最终会返回一个数据Array
    val stuArr: Array[String] = stuRDD.take(10)
    stuArr.foreach(println) // 这里的foreach是Scala中数据的方法，并不是Spark中的算子

    // collect：可以将RDD中的所有数据转换成数组Array，使用时需要注意数据量的大小
    val stuArr02: Array[String] = stuRDD.collect()

    // collectAsMap：将KV格式的RDD直接转换成本地集合Map
    // 也可以通过collect算子先得到Array再to成想要的集合类型
    val stuMap: collection.Map[String, String] = stuRDD.map(line => {
      val splits: Array[String] = line.split(",")
      (splits(0), splits(1))
    }).collectAsMap()

    // reduceByKeyLocally：进行分组统计后将结果转换成本地集合Map
    val clazzCntMap: collection.Map[String, Int] = stuRDD.map(line => {
      val splits: Array[String] = line.split(",")
      (splits(4), 1)
    }).reduceByKeyLocally(_ + _)

    // 等价于
    val clazzCntMap02: collection.Map[String, Int] = stuRDD.map(line => {
      val splits: Array[String] = line.split(",")
      (splits(4), 1)
    }).reduceByKey(_ + _).collectAsMap()

    // lookup：需要作用在KV格式的RDD，可以对指定的K进行查找出所有的Value最终构建成Seq集合
    val lookupSeq: Seq[Int] = stuRDD.map(line => {
      val splits: Array[String] = line.split(",")
      (splits(4), 1)
    }).lookup("文科四班")
    println(lookupSeq)

    // top：可以取出RDD中的前N条数据
    val topArr: Array[String] = stuRDD.top(10)
  }

}
